package cn.clx.ymcc.rocketmq;

import cn.clx.ymcc.domain.PayOrder;
import cn.clx.ymcc.dto.PayDto;
import cn.clx.ymcc.service.IPayOrderService;
import cn.clx.ymcc.util.AssertUtil;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.nio.charset.StandardCharsets;
import java.util.Date;

/**
 * 课程支付消费者
 */
@Slf4j
@Component
@RocketMQMessageListener(topic = "course-order-topic",
        selectorExpression="course-order-tags"	//tags
                ,consumerGroup = "service-order-course-pay-consumer")
public class CoursePayConsumer implements RocketMQListener<MessageExt> {

    @Autowired
    private IPayOrderService payOrderService;
    /**
     * 消费消息要保证消息的幂等性
     * @param message
     */
    @Override
    @Transactional
    public void onMessage(MessageExt message) {
        byte[] body = message.getBody();
        String paydtoTem = new String(body, StandardCharsets.UTF_8);
        PayDto payDto = JSONObject.parseObject(paydtoTem, PayDto.class);
        //消费消息方 报错就意味着消费失败，会重新进入队列，进行重新消费，直到mq对消息进行清扫
        AssertUtil.isNotNull(payDto,"消费消息失败");
        //校验一下是否已经被消费了
        PayOrder payOrder=payOrderService.checkPayOrder(payDto.getOrderNo());
        if(payOrder!=null){
            //已经被消费过了，直接进行签收
            return;
        }
        payOrder = new PayOrder();
        BeanUtils.copyProperties(payDto,payOrder);
        payOrder.setCreateTime(new Date());
        payOrder.setPayStatus(PayOrder.STATE_WAIT_PAY);
        payOrderService.insert(payOrder);
    }
}